package p.ithorns.integration.zk.lock;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * ZkConfWatcher
 * Zk 模拟配置中心
 *
 * @author ithorns
 * @version 1.0
 * @date 2019/9/21 22:30
 */
@Slf4j
public class ZkLockWatcher implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback {

    private final static String ROOT_DIR = "/";

    private final static String LOCK_PATH = "/lock";

    private ZooKeeper zk;

    private final Object ctx = "ZKLOCK";

    private CountDownLatch cdl = new CountDownLatch(1);

    private String threadName;

    private String pathName;

    public ZkLockWatcher(ZooKeeper zk, String threadName) {
        this.zk = zk;
        this.threadName = threadName;
    }

    @Override
    public void process(WatchedEvent event) {
        log.info("ZkLockWatcher ===========> event: {}", event.toString());
        Event.EventType type = event.getType();
        switch (type) {
            case NodeDeleted:
                // 监控上一个子节点的删除事件， 触发过程4
                zk.getChildren(ROOT_DIR, false, this, this.ctx);
                break;
        }
    }

    /**
     * 加锁
     * @author 2019/9/22-18:15 by Ht.L
     */
    public void trylock() {
        try {
            // 1. 创建节点，触发过程3
            zk.create(LOCK_PATH, threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, ctx);
            // 2. 在创建成功之前需要阻塞, 节点创建成功之后解除阻塞。
            cdl.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 临时节点创建 事件回调
     * @author 2019/9/22-16:50 by Ht.L
     * @param  rc, path, ctx, name
     */
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        // 3. 收到并处理节点创建事件，开始：getChildren产生子节点事件，触发过程4
        if (name != null) {
            log.info("=============> {} -create ephemeral node at path:{}, name: [{}]", threadName, path, name);
            // 当前创建的节点名称
            this.pathName = name;
            zk.getChildren(ROOT_DIR, false, this, ctx);
        }
    }

    /**
     * 子节点事件回调
     * @author 2019/9/21-23:22 by Ht.L
     * @param rc, path, ctx, children
     */
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {

        // 4. 收到并处理子节点事件
        Collections.sort(children);
        int index = children.indexOf(pathName.substring(1));
        if (index == 0) {
            // 如果是第一个节点
            log.info("{}: I am first", threadName);

            // TODO 可重入
            try {
                zk.setData(ROOT_DIR, threadName.getBytes(), -1);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 5. 节点创建成功 解除阻塞
            cdl.countDown();
        } else {
            // watch 上一个节点, 如果上一个节点删除了，产生删除事件，回调过程4
            try {
                zk.exists(ROOT_DIR + children.get(index - 1), this);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 解锁
     * @author 2019/9/22-18:15 by Ht.L
     */
    public void unlock() {
        try {
            zk.delete(pathName, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

}
